/* 
 * File:   JServerMaster.cpp
 * Author: johnneyshi
 * 
 * Created on 2013年1月27日, 下午8:16
 */

#include <signal.h>


#include "JServerMaster.h"
#include <arpa/inet.h>
#include <iostream>
#include <string>
#include <string.h>
#include <stdio.h>
#include "JException.h"
#include "JClient.h"
#include <fstream>
#include <boost/property_tree/ptree.hpp>
#include <boost/property_tree/xml_parser.hpp>
#include <boost/foreach.hpp>
#include "JLog.h"
#include "JDefine.h"
#include <sys/epoll.h>
#include "JEpoll.h"

using namespace std;

pthread_mutex_t needDealFdsLock = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t toWorkerFdsLock = PTHREAD_MUTEX_INITIALIZER;

//std::queue<int> adapterConnFdQueue;

JServerMaster::JServerMaster()
{
}

//JServerMaster::JServerMaster(const JServerMaster& orig) : JServer(orig)
//{
//}

JServerMaster::~JServerMaster() 
{
}

void JServerMaster::Init(std::string _configFileName)
{
    ReadConfig(_configFileName);
}

void JServerMaster::Start()
{
    for(int i = 0;i < masterWorkerAdapters.size(); ++i)
    {
        AddWorker(masterWorkerAdapters[i]);
    }
    JServer::Start();
}

void JServerMaster::AddWorker(MasterWorkerAdapter& adapter)
{
    JClient* pClient = adapter.pClient;
    JServerWorker* pWorker = adapter.pWorker;
    JLog::Log(JLogMessage,"Add worker, client IP(%s) port(%d), worker IP(%s) port(%d)\n",
            pClient->GetClientIP().c_str(),pClient->GetClientPort(),pWorker->GetIP().c_str(),pWorker->GetPort());
    int pipe1[2],pipe2[2];
    pid_t childPid;
    pipe(pipe1);
    pipe(pipe2);
    if((childPid = fork()) == 0)
    {
        close(pipe1[0]);
        close(pipe2[1]);
        //pipe1[1] write; pipe2[0] read
        MasterWorkerAdapterMain(pipe2[0],pipe1[1],adapter);
        
        while(true)
        {
            sleep(30000);
        }
        cout << "can't print" << endl;
        //char buff[200];
        //read(connFd,buff,200);
        //cout << buff << endl;
    }
    close(pipe1[1]);
    close(pipe2[0]);
    //pipe1[0] read; pipe2[1] write
    //this->readPipes.push_back(pipe1[0]);
    //this->writePipes.push_back(pipe2[1]);
    adapter.readPipe = pipe1[0];
    adapter.writePipe = pipe2[1];
}

void JServerMaster::ReadConfig(std::string _configFileName)
{
    using boost::property_tree::ptree;
    using namespace boost::property_tree::xml_parser;
    ptree pt;
    read_xml(_configFileName, pt);
    std::string masterIP = pt.get<std::string>("ServerMaster.IP");
    int masterPort = pt.get<int>("ServerMaster.Port");
    JServer::Init(masterIP,masterPort);
    
    BOOST_FOREACH(ptree::value_type &v,pt.get_child("ServerMaster.ServerWorkers"))
    {
        MasterWorkerAdapter adapter;
        std::string clientIP = v.second.get<std::string>("Client.IP","");
        int clientPort = v.second.get<int>("Client.Port",0);
        JClient* pClient = new JClient(clientIP,clientPort);
        //this->pClients.push_back(pClient);
        adapter.pClient = pClient;
        
        std::string ip = v.second.get<std::string>("Server.IP");
        int port = v.second.get<int>("Server.Port");
        int weight = v.second.get<int>("Weight");
        int portCount = v.second.get<int>("PortCount");
        JServerWorker* pWorker = new JServerWorker;
        pWorker->Init(ip,port,weight);
        //this->pWorkers.push_back(pWorker);
        //this->portCount.push_back(portCount);
        adapter.pWorker = pWorker;
        adapter.portCount = portCount;
        this->masterWorkerAdapters.push_back(adapter);
    }
}

//把接受到的消息，加上fd，然后转发给其中的一个worker
void JServerMaster::DealEpollFd(int fd)
{
//    int workIndex = AssignWork();
////    JBuffer* buff = GetAvailableBuffer();
////    buff->Put((char*)&fd,4);
//    char b[1024];///////////////////////////这里可以优化，如果不用b，直接把read到的数据放进buff中，会快
//    int n = 0;
//    while(true)
//    {
//        bzero(b,1024);
//        n = read(fd,b,1024);
//        if(n <= 0)
//        {
//            break;
//        }
//        write(this->writePipes[workIndex],b,n);
//    }
}

//计算公式：现在负载量=当前正在处理的工作/权重，选择现在负载量最小的进行分配
int JServerMaster::AssignWork()
{
    float minCapacity = numeric_limits<float>::max();
    int minIndex = -1;
    for(int i = 0;i < this->masterWorkerAdapters.size(); ++i)
    {
        float f = (float)(masterWorkerAdapters[i].pWorker->GetCurrentWorkCount()) / masterWorkerAdapters[i].pWorker->GetWeight();
        if(minCapacity > f)
        {
            minCapacity = f;
            minIndex = i;
        }
    }
    return minIndex;
}



/*这里需要一个机制，因为read函数是如果在缓冲中有数据的话，就会一直读；比如对端连续write了两次数据，
 * 而在read的时候，如果数组大的话，可以一次性读完这两次write的数据；read函数不会意识到有两次write,
 * 所以这里在接受到的数据前面，必须要有一个指示接下来接受到数据的大小。
 * 鉴于：
 * int最大：2147483647
 * long最大：9223372036854775807
 * 所以加一个4字节的大小就够了
 * 
 * 
 * 模式是Adapter和客户通信，先用JBuffer接受来自客户的所有数据，然后由adapter转发给worker
 * 好处：如果不这样，那么在客户和adapter通信时，就把数据传给worker，由于客户到adapter之间传输比较慢，
 * 而adapter到worker之间是内部网络，速度很快，这样就浪费了网速。这叫做异步传输。
 */
void JServerMaster::AdapterTransmitMsg(int readPipe, int writePipe, int toMasterConnFd,int toWorkerConnFd)
{
    JBuffer* pBuff = GetAvailableBuffer();
    int n = 0,needReadCount = 0;
    JEpoll epoll;
    epoll.Create(3);
    epoll.AddFd(readPipe);
    epoll.AddFd(toMasterConnFd);
    epoll.AddFd(toWorkerConnFd);
    std::vector<int> sockFds = epoll.Wait();
    for(int i = 0;i < sockFds.size(); ++i)
    {
        switch(sockFds[i])
        {
            case readPipe:
                break;
             
            //这是从客户端发过来的信息，操作：把接受到的信息缓存，缓存完毕，发给toWorkerConnFd
            case toMasterConnFd:
                //首先读取数据块的大小
                unsigned int size = 0;
                read(toMasterConnFd,(void*)&size,SIZEOF_UNSIGNED_INT);
                pBuff->Put((char*)&size,SIZEOF_UNSIGNED_INT);
                char buff[BUFF_SIZE];
                bzero(buff,BUFF_SIZE);
                
                while(size > 0)
                {
                    if(size > BUFF_SIZE)
                    {
                        needReadCount = BUFF_SIZE;
                    }
                    else
                    {
                        needReadCount = size;
                    }
                    n = read(toMasterConnFd,buff,needReadCount);
                    size -= n;
                    pBuff->Put(buff,n);
                    //这里不许要用下面语句，因为是根据n来写道pBuff中的
                    //bzero(buff,BUFF_SIZE);
                }
                //接受完数据，向worker转发
                while((n = pBuff->Get(buff,BUFF_SIZE)) > 0)
                {
                    write(toWorkerConnFd,buff,n);
                }
                break;
            //这是worker端发来的信息，由于接受的速度比较快，接受到数据后，不缓存，直接发送给toMasterConnFd
            case toWorkerConnFd:
                
                break;
        }
    }
    
    
    
    

}

void JServerMaster::DealNewConnfd(int newFd)
{
    int workIndex = AssignWork();
    //把得到的描述符传给下标为workIndex的进程
    write(newFd,(void*)&workIndex,sizeof(int));
    close(newFd);
}

/*Adapters是用于从队列中取出needDealFds,然后用缓存，得到所有的数据后，然后从ClientAdapters中选取一个
 * ClientAdapter，把数据发送给worker，然后接受来自worker处理完的数据，不需要缓存，直接传送给客户端，最后断开连接
 */
void* AdapterThread(void* arg)
{
    AdapterThreadArg* pArg = (AdapterThreadArg*)arg;
    JBuffer* pBuff = pArg->pMaster->GetAvailableBuffer();
    char buff[BUFF_SIZE];
    int needReadCount = 0,n = 0;
    int toMasterConnFd = -1;
    int toWorkerConnFd = -1,toWorkerConnFdIndex = -1;
    //设置这个是为了，如果客户端和服务器建立连接，如果要传输多次信息，但是不断开连接
    bool isCycleEnd = true;
    while(true)
    {
        if(isCycleEnd == true)
        {
            //从needDealFds中取fd
            toMasterConnFd = -1;
            while(true)
            {
                pthread_mutex_lock(&needDealFdsLock);
                if(!pArg->pMaster->needDealFds.empty())//////////////////////////////这里需要优化，不能
                {
                    toMasterConnFd = pArg->pMaster->needDealFds.front();//从共享文件中得到数据
                    pArg->pMaster->needDealFds.pop();
                }
                pthread_mutex_unlock(&needDealFdsLock);
                if(toMasterConnFd == -1)
                {
                    sleep(10);
                }
                else
                {
                    break;
                }
            }
        }
        //缓存数据
        //首先先把toMasterConnFd存入数据中，因为在返回数据的时候，还要用到toMasterConnFd,
        //问题：如果返回时，toMasterConnFd已经连接中断……////////////////////////////////////////////////////
        //pBuff->Put((char*)&toMasterConnFd,SIZEOF_INT);
        //读取数据块的大小
        unsigned int size = 0;
        n = read(toMasterConnFd,(void*)&size,SIZEOF_UNSIGNED_INT);
        if(n < 0 && errno == EINTR)/////////////////////////////////////////////////////
        //当客户端向服务器写一个size=0的数据时，代表连接结束
        if(size == 0)
        {
            isCycleEnd = true;
            close(toMasterConnFd);
            continue;
        }
        pBuff->Put((char*)&size,SIZEOF_UNSIGNED_INT);
        bzero(buff,BUFF_SIZE);
        while(size > 0)
        {
            if(size > BUFF_SIZE)
            {
                needReadCount = BUFF_SIZE;
            }
            else
            {
                needReadCount = size;
            }
            n = read(toMasterConnFd,buff,needReadCount);
            size -= n;
            pBuff->Put(buff,n);
            //这里不许要用下面语句，因为是根据n来写道pBuff中的
            //bzero(buff,BUFF_SIZE);
        }
        /*接受完数据,选取toWorkerFd，这里必须要找到toWorkerFd,如果没有，表示worker已经用完，那么需要循环等待，
         * 直到得到
         */
        toWorkerConnFd = -1;
        
        //如果没有得到toWorkerFd,不让循环退出
        while(true)
        {
            pthread_mutex_lock(&toWorkerFdsLock);
            for(int i = 0;i < pArg->pMaster->toWorkerFds.size(); ++i)
            {
                if(pArg->pMaster->toWorkerFds[i] > 0)
                {
                    toWorkerConnFd = pArg->pMaster->toWorkerFds[i];
                    pArg->pMaster->toWorkerFds[i] = -1 * toWorkerConnFd;
                    toWorkerConnFdIndex = i;
                    break;
                }
            }
            pthread_mutex_unlock(&toWorkerFdsLock);
            if(toWorkerConnFd == -1)
            {
                sleep(10);
            }
            else
            {
                break;
            }
        }
        //得到toWorkerFd,向toWorkerConnFd发送数据
        while((n = pBuff->Get(buff,BUFF_SIZE)) > 0)
        {
             write(toWorkerConnFd,buff,n);
        }
        //从toWorkerFd中读取处理后的数据,不需要缓存，直接传给客户端
        read(toWorkerConnFd,(void*)&size,SIZEOF_UNSIGNED_INT);
        write(toMasterConnFd,(void*)&size,SIZEOF_UNSIGNED_INT);
        while(size > 0)
        {
            if(size > BUFF_SIZE)
            {
                needReadCount = BUFF_SIZE;
            }
            else
            {
                needReadCount = size;
            }
            n = read(toWorkerConnFd,buff,needReadCount);
            size -= n;
            write(toMasterConnFd,buff,n);
        }
        //当数据写完，就应该把toWorkerConnFd的状态置为可以处理
        pthread_mutex_lock(&toWorkerFdsLock);
        pArg->pMaster->toWorkerFds[toWorkerConnFdIndex] = -1 * toMasterConnFd;
        pthread_mutex_unlock(&toWorkerFdsLock);
        //一个处理周期结束.
    }
}


void* ClientAdapterThread(void* arg)
{
    AdapterThreadArg* pArg = (AdapterThreadArg*)arg;
    JClient client(pArg->pAdapter->pClient->GetClientIP(),pArg->port);
    client.Connect(pArg->pAdapter->pWorker->GetIP(),pArg->pAdapter->pWorker->GetPort());
    int toWorkerConnFd = client.GetConnFd();
    pArg->pMaster->toWorkerFds.push_back(toWorkerConnFd);
    //char msg[200];
    //bzero(msg,200);
    JLog::Log(JLogMessage,"Connect to worker success. Worker IP:%s, worker port:%d\n",pArg->pAdapter->pWorker->GetIP().c_str(),pArg->pAdapter->pWorker->GetPort());
}


/*
 * Adapters是用于从队列中取出needDealFds,然后用缓存，得到所有的数据后，然后从ClientAdapters中选取一个
 * ClientAdapter，把数据发送给worker。
 * ClientAdapter的用于发送数据给worker和接受来自worker的数据。
 */
void JServerMaster::MasterWorkerAdapterMain(int readFd,int writeFd,MasterWorkerAdapter& adapter)
{
    //设置线程最高优先级
//    pthread_attr_t attr;
    struct sched_param param;
//    pthread_attr_init(&attr);
//    pthread_attr_setschedpolicy(&attr,SCHED_RR);
    param.__sched_priority = 99;
//    pthread_attr_setschedparam(&attr,param);
    sched_setscheduler(getpid(),SCHED_RR,&param);
    /////////
    pthread_t tid;
//    //错：多线程，创建ClientAdapters////////////////////////////////////这里不许要多线程，只需要循环创建portCount个client就行了
    for(int i = 0;i < adapter.portCount; ++i)
    {
        AdapterThreadArg arg;
        arg.pAdapter = &adapter;
        arg.pMaster = this;
        arg.childReadPipe = readFd;
        arg.childWritePipe = writeFd;
        arg.port = adapter.pClient->GetClientPort() + i;
        ClientAdapterThread((void*)&arg);
    }
    //多线程，创建2倍的Adapters，暂时定Adapters是ClientAdapters的2倍，因为Adapters要比ClientAdapters多，才会有意义
    for(int i = 0;i < 2 * adapter.portCount; ++i)
    {
        AdapterThreadArg arg;
        arg.pAdapter = &adapter;
        arg.pMaster = this;
        arg.childReadPipe = readFd;
        arg.childWritePipe = writeFd;
        pthread_create(&tid,NULL,&AdapterTransmitMsg,(void*)&arg);
    }
    
    //把接受到的要处理的connFd都放入一个队列中，各个线程从中取connFd;
    while(true)
    {
        int newFd = -1;//
        read(readFd,(void*)&newFd,sizeof(int));
        pthread_mutex_lock(&needDealFdsLock);////////////////////加锁，这里有疑问，可能获得不到锁，就会读不到队列.
        needDealFds.push(newFd);
        pthread_mutex_unlock(&needDealFdsLock);
    }
}

